package com.fy.service.imbusiness.controller;

import com.baomidou.mybatisplus.mapper.EntityWrapper;
import com.fy.common.constant.CmdConstants;
import com.fy.common.dto.MessageGetDto;
import com.fy.common.dto.MessageRecallDelDto;
import com.fy.common.entity.SkImChatSession;
import com.fy.common.enums.SkImCommonStatusEnum;
import com.fy.common.model.im.MessageModel;
import com.fy.common.util.ImUtil;
import com.fy.common.util.TokenUtil;
import com.fy.common.vo.MessageVo;
import com.fy.framework.common.model.R;
import com.fy.framework.component.config.LocaleMessage;
import com.fy.framework.component.config.RedissonComponet;
import com.fy.framework.component.config.TenantContextHolder;
import com.fy.service.service.ImBrokerService;
import com.fy.service.service.SkImChatSessionService;
import com.fy.service.service.SkImUserInfoService;
import com.google.common.collect.Iterators;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.redisson.api.RMapCache;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.io.IOException;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.util.*;

import static com.fy.common.util.ImUtil.chatSessionKeyGen;

/**
 * @author zhshuo
 * @date 2021/12/14 11:42
 */
@Slf4j
@RestController
@Api(tags = "消息接口")
@RequestMapping("/message")
public class SkImMessageController {

    public static final byte[] MESSAGE_BYTES = Bytes.toBytes("message");
    public static final byte[] CMD_BYTES = Bytes.toBytes("cmd");
    public static final byte[] SEND_TIME_BYTES = Bytes.toBytes("sendTime");
    public static final String PIG_SINGLE = "pig_single_";

    @Autowired
    LocaleMessage localeMessage;

    @Autowired
    RedissonComponet redissonComponet;

    @Autowired
    Connection hbaseConnection;

    @Autowired
    SkImUserInfoService userInfoService;

    @Autowired
    SkImChatSessionService chatSessionService;

    @ApiOperation("查询用户的消息列表")
    @PostMapping("get")
    public R<?> messageGet(@RequestBody MessageGetDto messageGetDto) throws IOException {
        Long getDtoBeginTime = messageGetDto.getBeginTime();
        if(getDtoBeginTime == null || getDtoBeginTime.equals(0L)){
            return R.failedAlter(localeMessage.getMessage("im.common.parameterError"));
        }

        String sessionId = messageGetDto.getSessionId();
        String chatSessionKey = null;
        if (sessionId.startsWith(ImUtil.FRIEND_CHAT_SESSION_PREFIX)) {
            chatSessionKey = ImUtil.chatSessionIdGenFriend(TokenUtil.getUserId(), messageGetDto.getTargetId());
        } else if (sessionId.startsWith(ImUtil.GROUP_CHAT_SESSION_PREFIX)) {
            chatSessionKey = ImUtil.chatSessionIdGenGroup(TokenUtil.getUserId(), messageGetDto.getTargetId());
        } else {
            return R.failedAlter(localeMessage.getMessage("im.common.parameterError"));
        }
        RMapCache<String, Object> mapCache = redissonComponet.getMap(chatSessionKeyGen(TokenUtil.getUserId().toString()));
        SkImChatSession session;
        if (mapCache.containsKey(chatSessionKey)) {
            session = (SkImChatSession) mapCache.get(chatSessionKey);
        } else {
            session = chatSessionService.selectOne(new EntityWrapper<SkImChatSession>().eq("session_id", chatSessionKey).eq("status", SkImCommonStatusEnum.NORAML.getStatus()));
        }

        if (Objects.isNull(session)) {
            return R.failedAlter(localeMessage.getMessage("im.common.parameterError"));
        }
        PrefixFilter prefixFilter = new PrefixFilter(Bytes.toBytes(new StringBuilder(PIG_SINGLE).append(TokenUtil.getUserId()).append("_").append(messageGetDto.getTargetId()).toString()));
        List<MessageVo> messageVoList = new ArrayList<>();
        Table table = hbaseConnection.getTable(TableName.valueOf("message:message"));
        Scan scaWithNoRowKey = new Scan();
        FilterList allFilters = new FilterList(FilterList.Operator.MUST_PASS_ALL);
        SingleColumnValueFilter columnValueFilter = new SingleColumnValueFilter(MESSAGE_BYTES, CMD_BYTES, CompareOperator.EQUAL, Bytes.toBytes(CmdConstants.RESPONSE_SINGLE_CHAT));
        SingleColumnValueFilter endTimeFilter = new SingleColumnValueFilter(MESSAGE_BYTES, SEND_TIME_BYTES, CompareOperator.LESS_OR_EQUAL, Bytes.toBytes(getDtoBeginTime.toString()));
        RowFilter rowFilter = new RowFilter(CompareOperator.EQUAL, new BinaryPrefixComparator(Bytes.toBytes(new StringBuilder(PIG_SINGLE).append(TokenUtil.getUserId()).append("_").append(messageGetDto.getTargetId()).toString())));
        allFilters.addFilter(columnValueFilter);
        allFilters.addFilter(endTimeFilter);
        allFilters.addFilter(rowFilter);
        scaWithNoRowKey.setFilter(allFilters);
        scaWithNoRowKey.setReversed(true);
        scaWithNoRowKey.setLimit(20);
        ResultScanner scannerFirst = table.getScanner(scaWithNoRowKey);

        Iterator<Result> iterator = scannerFirst.iterator();
        if(!iterator.hasNext()){
            return R.success(null);
        }

        long beginTime = Long.parseLong(Bytes.toString(Iterators.getLast(iterator).getValue(MESSAGE_BYTES, Bytes.toBytes("sendTime"))));
        if (Objects.nonNull(session.getLastDeleteTime())) {
            long time = session.getLastDeleteTime().getTime();
            beginTime = Math.max(beginTime, time);
        }

        FilterList allFiltersData = new FilterList(FilterList.Operator.MUST_PASS_ALL);
        SingleColumnValueFilter dataBegin = new SingleColumnValueFilter(MESSAGE_BYTES, SEND_TIME_BYTES, CompareOperator.GREATER_OR_EQUAL, Bytes.toBytes(String.valueOf(beginTime)));
        SingleColumnValueFilter dataEnd = new SingleColumnValueFilter(MESSAGE_BYTES, SEND_TIME_BYTES, CompareOperator.LESS_OR_EQUAL, Bytes.toBytes(String.valueOf(getDtoBeginTime)));
        Scan scanData = new Scan();
        allFiltersData.addFilter(prefixFilter);
        allFiltersData.addFilter(dataBegin);
        allFiltersData.addFilter(dataEnd);
        scanData.setFilter(allFiltersData);
        ResultScanner scannerData = table.getScanner(scanData);
        Iterator<Result> scannerDataIterator = scannerData.iterator();
        while (scannerDataIterator.hasNext()) {
            Result res = scannerDataIterator.next();
            messageVoList.add(new MessageVo()
                    .setRowKey(Bytes.toString(res.getRow()))
                    .setCmd(Bytes.toString(res.getValue(MESSAGE_BYTES, Bytes.toBytes("cmd"))))
                    .setDateStr(Bytes.toString(res.getValue(MESSAGE_BYTES, Bytes.toBytes("dateStr"))))
                    .setContent(Bytes.toString(res.getValue(MESSAGE_BYTES, Bytes.toBytes("content"))))
                    .setContentType(Bytes.toString(res.getValue(MESSAGE_BYTES, Bytes.toBytes("contentType"))))
                    .setFrom(Bytes.toString(res.getValue(MESSAGE_BYTES, Bytes.toBytes("from"))))
                    .setMessageId(Bytes.toString(res.getValue(MESSAGE_BYTES, Bytes.toBytes("messageId"))))
                    .setSendTime(Bytes.toString(res.getValue(MESSAGE_BYTES, Bytes.toBytes("sendTime"))))
                    .setSize(Bytes.toString(res.getValue(MESSAGE_BYTES, Bytes.toBytes("size"))))
                    .setTo(Bytes.toString(res.getValue(MESSAGE_BYTES, Bytes.toBytes("to")))));
        }
        scannerFirst.close();
        scannerData.close();
        return R.success(messageVoList);
    }


}
